package future;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import util.Util;

import java.math.BigDecimal;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @author yding5
 */
public class FutureVsReactor {

    private static Map<Long, String> NAME_MAP;

    private static Map<Long, BigDecimal> PRICE_MAP;

    @Before
    public void setup() {
        NAME_MAP = new HashMap() {
            {
                put(1L, "DIOR");
                put(2L, "LANCOME");
                put(3L, "CLINIQUE");
                put(4L, "SEPHORA");
                put(5L, "YSL");
            }
        };

        PRICE_MAP = new HashMap() {
            {
                put(1L, new BigDecimal("385.00"));
                put(2L, new BigDecimal("312.00"));
                put(3L, new BigDecimal("760.00"));
                put(4L, new BigDecimal("288.00"));
                put(5L, new BigDecimal("315.00"));
            }
        };
    }

    @After
    public void cleanup() {
        NAME_MAP.clear();
        PRICE_MAP.clear();
    }

    @Test
    public void test1() {
        Util.timeWatch();
        CompletableFuture<List<Long>> ids = getIdList();
        CompletableFuture<List<String>> result = ids.thenCompose(list -> {
            List<CompletableFuture<String>> combineTask = list.stream().map(id -> {
                CompletableFuture<String> nameTask = getNameFuture(id);
                CompletableFuture<BigDecimal> priceTask = getPriceFuture(id);
                return nameTask.thenCombine(priceTask, (name, price) -> "brand " + name + " is " + price);
            }).collect(Collectors.toList());

            CompletableFuture<String>[] combinationArray =
                combineTask.toArray(new CompletableFuture[combineTask.size()]);
            CompletableFuture<Void> allDone = CompletableFuture.allOf(combinationArray);
            return allDone.thenApply(v -> combineTask.stream().map(CompletableFuture::join).collect(Collectors.toList()));
        });
        System.out.println(result.join());
        Util.timeWatch();
    }

    @Test
    public void test2() {
        Util.timeWatch();
        Flux<Long> ids = reactGetIdList();
        Flux<String> combinations = ids.flatMap(id -> {
            Mono<String> nameTask = reactGetName(id);
            Mono<BigDecimal> priceTask = reactGetPrice(id);
            return nameTask.zipWith(priceTask, (name, price) -> "brand " + name + " is " + price);
        });
        Mono<List<String>> result = combinations.collectList();
        System.out.println(result.block());
        Util.timeWatch();
    }

    private static CompletableFuture<BigDecimal> getPriceFuture(Long i) {
        return CompletableFuture.supplyAsync(() -> getPriceById(i));
    }

    private static CompletableFuture<String> getNameFuture(Long i) {
        return CompletableFuture.supplyAsync(() -> getNameById(i));
    }

    private static CompletableFuture<List<Long>> getIdList() {
        return CompletableFuture.supplyAsync(() -> {
            Util.sleep(100);
            Util.print("getIdList");
            return Arrays.asList(1L, 2L, 3L, 4L, 5L);
        });
    }

    private static String getNameById(Long id) {
        Util.sleep(100);
        Util.print("getNameById id: " + id);
        return NAME_MAP.get(id);
    }

    private static BigDecimal getPriceById(Long id) {
        Util.sleep(150);
        Util.print("getPriceById id: " + id);
        return PRICE_MAP.get(id);
    }

    private static Mono<BigDecimal> reactGetPrice(Long id) {
        return Mono.fromCallable(() -> getPriceById(id)).subscribeOn(Schedulers.elastic());
    }

    private static Mono<String> reactGetName(Long id) {
        return Mono.fromCallable(() -> getNameById(id)).subscribeOn(Schedulers.elastic());
    }

    private static Flux<Long> reactGetIdList() {
        return Flux.create((Consumer<? super FluxSink<Long>>)fluxSink -> {
            for (Long i = 1L; i <= 5L; i++) {
                fluxSink.next(i);
            }
            fluxSink.complete();
            Util.sleep(100);
            Util.print("getIdList");
        });
    }
}
